package com.yl.util;

import cn.hutool.core.util.StrUtil;
import com.yl.constant.Const;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.connectors.influxdb.InfluxDBConfig;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

public class InfluxUtil {

    /**
     * 初始化influxdb客户端配置
     */
    public static InfluxDB connect(InfluxDBConfig influxDBConfig) {
        InfluxDB influxDBClient = InfluxDBFactory.connect(influxDBConfig.getUrl(), influxDBConfig.getUsername(), influxDBConfig.getPassword());
        if (!influxDBClient.databaseExists(influxDBConfig.getDatabase())) {
            if (!influxDBConfig.isCreateDatabase()) {
                throw new RuntimeException("This " + influxDBConfig.getDatabase() + " database does not exist!");
            }
            influxDBClient.createDatabase(influxDBConfig.getDatabase());
        }
        influxDBClient.setDatabase(influxDBConfig.getDatabase());
        if (influxDBConfig.getBatchActions() > 0) {
            influxDBClient.enableBatch(influxDBConfig.getBatchActions(), influxDBConfig.getFlushDuration(), influxDBConfig.getFlushDurationTimeUnit());
        }
        if (influxDBConfig.isEnableGzip()) {
            influxDBClient.enableGzip();
        }
        return influxDBClient;
    }

    /**
     * 执行 influxdb sql
     */
    public static List<QueryResult.Series> executeSql(InfluxDB influxDB, String db, String sql) {
        List<QueryResult.Series> res = new ArrayList<>();
        QueryResult query = influxDB.query(new Query(sql, db));
        Optional.ofNullable(query.getResults().get(0).getSeries())
                .ifPresent(res::addAll);
        return res;
    }

}
